package com.fxkj.cloud.drms.infra.utils;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Nullable;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * 数据处理器
 * @param <T> 异步读写任务
 * @param <F> 异步任务结果
 * @param <V> 异步任务结果类型
 */
@Slf4j
public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extends ListenableFuture<V>, V> implements BufferedRateExecutor<T, F> {
    public static final String CONCURRENCY_LEVEL = "currBuffer";
    private final long maxWaitTime;
    private final long pollMs;
    private final BlockingQueue<AsyncTaskContext<T, V>> queue;
    private final ExecutorService dispatcherExecutor;
    private final ExecutorService callbackExecutor;
    private final ScheduledExecutorService timeoutExecutor;

    public AbstractBufferedRateExecutor(int queueLimit, int concurrencyLimit, long maxWaitTime, int dispatcherThreads,
                                        int callbackThreads, long pollMs
    ) {
        this.maxWaitTime = maxWaitTime;
        this.pollMs = pollMs;
        this.queue = new LinkedBlockingDeque<>(queueLimit);
        this.dispatcherExecutor = Executors.newFixedThreadPool(dispatcherThreads, FxThreadFactory.forName("nosql-" + getBufferName() + "-dispatcher"));
        this.callbackExecutor = FxExecutors.newWorkStealingPool(callbackThreads, "nosql-" + getBufferName() + "-callback");
        this.timeoutExecutor = Executors.newSingleThreadScheduledExecutor(FxThreadFactory.forName("nosql-" + getBufferName() + "-timeout"));
        for (int i = 0; i < dispatcherThreads; i++) {
            dispatcherExecutor.submit(this::dispatch);
        }
    }

    @Override
    public F submit(T task) {
        SettableFuture<V> settableFuture = create();
        F result = wrap(task, settableFuture);

        try {
            queue.add(new AsyncTaskContext<>(UUID.randomUUID(), task, settableFuture, System.currentTimeMillis()));
        } catch (IllegalStateException e) {
            settableFuture.setException(e);
        }
        return result;
    }

    public void stop() {
        if (dispatcherExecutor != null) {
            dispatcherExecutor.shutdownNow();
        }
        if (callbackExecutor != null) {
            callbackExecutor.shutdownNow();
        }
        if (timeoutExecutor != null) {
            timeoutExecutor.shutdownNow();
        }
    }

    protected abstract SettableFuture<V> create();

    protected abstract F wrap(T task, SettableFuture<V> future);

    protected abstract ListenableFuture<V> execute(AsyncTaskContext<T, V> taskCtx);

    public abstract String getBufferName();

    private void dispatch() {
        log.info("[{}] Buffered rate executor thread started", getBufferName());
        while (!Thread.interrupted()) {
            AsyncTaskContext<T, V> taskCtx = null;
            try {
                if (!queue.isEmpty()) {
                    taskCtx = queue.take();
                    final AsyncTaskContext<T, V> finalTaskCtx = taskCtx;
                    long timeout = finalTaskCtx.getCreateTime() + maxWaitTime - System.currentTimeMillis();
                    if (timeout > 0) {
                        ListenableFuture<V> result = execute(finalTaskCtx);
                        result = Futures.withTimeout(result, timeout, TimeUnit.MILLISECONDS, timeoutExecutor);
                        Futures.addCallback(result, new FutureCallback<V>() {
                            @Override
                            public void onSuccess(@Nullable V result) {
                                finalTaskCtx.getFuture().set(result);
                            }

                            @Override
                            public void onFailure(Throwable t) {
                                finalTaskCtx.getFuture().setException(t);
                                log.debug("[{}] Failed to execute task: {}", finalTaskCtx.getId(), finalTaskCtx.getTask(), t);
                            }
                        }, callbackExecutor);
                    } else {
                        taskCtx.getFuture().setException(new TimeoutException());
                    }
                }
            } catch (InterruptedException e) {
                break;
            } catch (Throwable e) {
                if (taskCtx != null) {
                    log.debug("[{}] Failed to execute task: {}", taskCtx.getId(), taskCtx, e);
                } else {
                    log.debug("Failed to queue task:", e);
                }
            }
        }
        log.info("[{}] Buffered rate executor thread stopped", getBufferName());
    }

}
